package com.pro.common.service.file.service.unit;

import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.obs.services.exception.ObsException;
import com.pro.common.api.R;
import com.pro.common.api.depencies.util.BeanMergeUtils;
import com.pro.common.api.model.ILoginInfo;
import com.pro.common.api.service.ITestService;
import com.pro.common.api.statics.user.file.storage.enums.EnumFileStorageBusiness;
import com.pro.common.api.statics.user.file.storage.model.dto.UploadFIleResultDTO;
import com.pro.common.api.statics.user.file.storage.service.IFileStorageService;
import com.pro.common.api.user.file.enums.EnumFileStatus;
import com.pro.common.api.user.file.model.db.UserFile;
import com.pro.common.api.user.file.model.db.UserFileNode;
import com.pro.common.api.user.file.model.db.UserUploadRecord;
import com.pro.common.file.storage.config.FileStorageProperties;
import com.pro.common.service.file.model.dto.FileUploadDTO;
import com.pro.common.service.file.model.mapstruct.FileUploadDTOMapper;
import com.pro.common.service.file.model.request.FileUploadBatchRequest;
import com.pro.common.service.file.model.request.UserUploadRecordRequest;
import com.pro.common.service.file.service.AdaptiveConcurrencyControl;
import com.pro.common.service.file.service.UserFileNodeService;
import com.pro.common.service.file.service.UserFileService;
//import com.pro.common.service.file.service.UserFolderService;
import com.pro.common.service.file.service.UserUploadRecordService;
import com.pro.common.service.file.util.sub.UserFilePathParam;
import com.pro.framework.api.file.storage.PresignedUrlResponse;
import com.pro.framework.api.structure.Tuple2;
import com.pro.framework.api.util.*;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.io.InputStream;
import java.net.SocketTimeoutException;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 文件聚合服务
 */
@Service
@Slf4j
public class UserFileStorageUnitService implements ITestService {
    @Autowired(required = false)
    private IFileStorageService storageService;
    @Autowired
    private UserFileUnitService userFileUnitService;
    @Autowired
    private UserFileService userFileService;
    @Autowired
    private UserFileNodeService userFileNodeService;
    @Autowired
    private FileUploadDTOMapper fileUploadDTOMapper;
    @Autowired
    private FileStorageProperties fileStorageProperties;
    @Autowired
    private UserUploadRecordService userUploadRecordService;
    @Autowired
    private ThreadPoolTaskExecutor executor;
    @Autowired
    private AdaptiveConcurrencyControl adaptiveConcurrencyControl;

    @Transactional(rollbackFor = Exception.class)
    public List<R<UserFileNode>> putObjectsCreateFileNode(List<UserFilePathParam> userFilePathRequests) {
        AssertUtil.notEmpty(storageService, "请先实现文件上传服务");


        String bucketName = fileStorageProperties.getBusinessBucketMap().get(EnumFileStorageBusiness.userFileStorage.name());

        // 1. 构造文件元信息（含父目录链）-----------
        List<Tuple2<UserFilePathParam, UserFileNode>> tuple2s = userFilePathRequests.stream()
                .map(req -> new Tuple2<>(req, userFileUnitService.buildFull(req, bucketName)))
                .toList();

        // 2. 执行上传
//        List<R<UploadFIleResultDTO>> results = tuple2s.stream().map(tuple2 -> {
//            UserFilePathParam request = tuple2.getT1();
//            UserFileNode fileNode = tuple2.getT2();
//            try {
//                R<UploadFIleResultDTO> rs = this.putObjectSafe(fileNode, request.getUrl());
//                Thread.sleep(100);
//                return rs;
//            } catch (Exception e) {
////                throw new RuntimeException(e);
//                return R.<UploadFIleResultDTO>fail(e.getMessage());
//            }
//        }).toList();
        int currentConcurrency = adaptiveConcurrencyControl.getConcurrency();
        List<R<UploadFIleResultDTO>> results = AsyncExecutorUtil.executeConcurrentlyPart(
                tuple2s,
                tuple2 -> {
                    UserFilePathParam request = tuple2.getT1();
                    UserFileNode fileNode = tuple2.getT2();
                    return this.putObjectSafe(fileNode, request.getUrl());
                },
                executor,
                currentConcurrency // 每批并发数
        );

        // 3. 组装上传结果
        List<R<UserFileNode>> rsList = CollUtils.forEach(tuple2s, results, (tuple2, result) -> {
            UserFileNode fileNode = tuple2.getT2();
            if (result.isSuccess()) {
                UploadFIleResultDTO data = result.getData();
                fileNode.setUrl(data.getUrl());
                fileNode.getUserFile().setEtag(data.getEtag());
                return R.ok(fileNode);
            } else {
                return R.fail(result.getMsg());
            }
        });

        // 4. 保存成功的文件元信息
        List<UserFileNode> successNodes = rsList.stream().filter(R::isSuccess).map(R::getData).toList();

        userFileUnitService.saveBatchFull(successNodes);
        return rsList;
    }


    private OkHttpClient httpClient = new OkHttpClient.Builder()
            .connectTimeout(10, TimeUnit.SECONDS)
            .writeTimeout(60, TimeUnit.SECONDS)  // 很重要：上传用的 write timeout
            .readTimeout(60, TimeUnit.SECONDS)
            .retryOnConnectionFailure(true)
            .build();

    private R<UploadFIleResultDTO> putObjectSafe(UserFileNode fileNode, String url) {
        UserFile userFile = fileNode.getUserFile();
        String bucket = userFile.getBucket();
        String storageKey = userFile.getStorageKey();
        if (storageKey.startsWith("/")) {
            storageKey = storageKey.substring(1);
        }

        return getUploadFIleResultDTOR(url, bucket, storageKey, fileNode.getSizeBytes());
    }

    @SneakyThrows
    private R<UploadFIleResultDTO> getUploadFIleResultDTOR(String url, String bucket, String storageKey, Long sizeBytes) {
        int maxRetry = 3;

        for (int attempt = 1; attempt <= maxRetry; attempt++) {
            long startTime = System.currentTimeMillis(); // 开始时间
            Response response = null;
            InputStream in = null;
            try {
                // 1. 下载源文件
                Request request = new Request.Builder().url(url).build();
                response = httpClient.newCall(request).execute();

                // 2. 判断 HTTP 状态码
                if (!response.isSuccessful()) {
                    int code = response.code();
                    // 5xx 可以重试
                    if (code >= 500) {
                        throw new IOException("服务器错误, httpCode=" + code);
                    } else {
                        // 4xx 不重试
                        adaptiveConcurrencyControl.recordFailure();
                        return R.fail("下载失败, httpCode=" + code);
                    }
                }

                // 3. 获取输入流，手动关闭
                in = response.body().byteStream();

                // 4. 上传到 OBS
                UploadFIleResultDTO rs = storageService.putObject(bucket, storageKey, in);

                // 上传成功
                adaptiveConcurrencyControl.recordSuccess();

                long endTime = System.currentTimeMillis();   // 结束时间
                long durationMs = endTime - startTime;      // 耗时毫秒
//                OBS={}
                log.info("✅ 上传成功: URL={} ->, 大小={} KB, 耗时={} ms",
                        url, null == sizeBytes ? -1L : (sizeBytes / 1024), durationMs);
                return R.ok(rs);

            } catch (SocketTimeoutException e) {
                adaptiveConcurrencyControl.recordFailure();
                log.warn("超时 (第{}次尝试) url={}", attempt, url, e);
                if (attempt < maxRetry) {
                    Thread.sleep(500L * attempt);
                    continue;
                }
                return R.fail("下载超时: " + e.getMessage());
            } catch (IOException e) {
                adaptiveConcurrencyControl.recordFailure();
                log.warn("IO异常 (第{}次尝试) url={}, message={}", attempt, url, e.getMessage());
                if (attempt < maxRetry) {
                    Thread.sleep(500L * attempt);
                    continue;
                }
                return R.fail("下载失败: " + e.getMessage());
            } catch (ObsException e) {
                adaptiveConcurrencyControl.recordFailure();
                log.error("OBS上传异常 (第{}次尝试) bucket={}, key={}, url={}, message={}", attempt, bucket, storageKey, url, e.getMessage(), e);
                // 判断是否可重试
                if (attempt < maxRetry && isRetryableObsException(e)) {
                    Thread.sleep((long) Math.pow(2, attempt) * 200L); // 200ms, 400ms, 800ms...
                    continue;
                }
                return R.fail("OBS上传失败: " + e.getMessage());

            } finally {
                try {
                    if (in != null) in.close();
                } catch (Exception ignored) {
                }
                try {
                    if (response != null) response.close();
                } catch (Exception ignored) {
                }
            }
        }

        return R.fail("上传失败: 达到最大重试次数");
    }

    /**
     * 判断 OBS 异常是否可重试
     */
    private boolean isRetryableObsException(ObsException e) {
        String code = e.getErrorCode();
        // 可重试的错误码：InternalError、RequestTimeout、ServiceUnavailable
        return "InternalError".equals(code)
                || "RequestTimeout".equals(code)
                || "ServiceUnavailable".equals(code);
    }


    @Override
    public void test() {


        List<String> urls = JSONUtil.toList("[" +
                "\"http://110.41.10.45:8001/images/2025/11/06/106043_3d9a8844f35c46f9a97420f8769b40a4.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/106259_94f75c21bb2c44d387f8834c5c5e9c35.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/106433_f54b3c4e49864b0692e6eb2f55b8923a.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/106667_8f1126b0213a4beebc2bfc14d21cee66.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/106834_7f199fd87ae74435ba4dcdca23ebf329.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/107033_048d7d72a1c24405b267c60cdaeca80b.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/107242_12b6dfbe5c5345caa45838d47163ffe9.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/107440_7ebf797db6a648f0a8c34739db632148.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/107637_3684647948604137b95b5b6428df2c52.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/107844_cc575638f6144ba48893b6bf2c3cd3ca.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/108626_7320f31fc17740d6b74bb728266829d8.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/108818_823a5d8351884d18ae72f6a63ed7c417.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/109046_c285680cca0049acbdeed770af7e96f2.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/109222_71801ffca96443f9a8ca958cecd5d9b6.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/109420_c1eba6a051674d74b8226cacb726bf11.jpg\"," +
                "\"\"," +
                "\"\"," +
                "\"\"]", String.class).stream().filter(StrUtil::isNotEmpty).toList();
        List<Tuple2<UserFilePathParam, UserFileNode>> tuple2s = new ArrayList<>();

        for (String url : urls) {
            // 构建 UserFilePathParam
            UserFilePathParam param = new UserFilePathParam(
                    "testModule", "测试模块", 123L, "batch-001", UUID.randomUUID().toString(),
                    url.substring(url.lastIndexOf("/") + 1), url, 123L
            );

            // 构建 UserFileNode 和 UserFile
            UserFile userFile = new UserFile();
            userFile.setBucket("upload-baoyi");
            userFile.setStorageKey("test-upload/" + param.getOriginalFilename());
//            userFile.setSizeBytes(0L); // 可选，测试用

            UserFileNode node = new UserFileNode();
            node.setUserFile(userFile);
            node.setNodeName(param.getOriginalFilename());

            // kotlin.Tuple2 或你自己实现的 Tuple2
            tuple2s.add(new Tuple2<>(param, node));
        }

        // 并发上传
        int currentConcurrency = adaptiveConcurrencyControl.getConcurrency();
        List<R<UploadFIleResultDTO>> results = AsyncExecutorUtil.executeConcurrentlyPart(
                tuple2s,
                tuple2 -> {
                    UserFilePathParam request = tuple2.getT1();
                    UserFileNode fileNode = tuple2.getT2();
                    return this.putObjectSafe(fileNode, request.getUrl());
                },
                executor,
                currentConcurrency // 每批并发数
        );

        // 输出结果
        for (int i = 0; i < tuple2s.size(); i++) {
            R<UploadFIleResultDTO> result = results.get(i);
            System.out.println("URL: " + tuple2s.get(i).getT1().getUrl());
            if (result.isSuccess()) {
                System.out.println("上传成功: " + result.getData().getUrl());
            } else {
                System.out.println("上传失败: " + result.getMsg());
            }
            System.out.println("-------------------------------------");
        }

//            executor.shutdown();
//            executor.awaitTermination(1, TimeUnit.MINUTES);
    }

    private void testUint() {
        // ====================== 配置区 ======================

        //  公网 http://110.41.10.45:8001
        //  局域网 http://172.31.5.219:8001
        List<String> fileUrls = JSONUtil.toList("[" +
                "\"http://110.41.10.45:8001/images/2025/11/06/106043_3d9a8844f35c46f9a97420f8769b40a4.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/106259_94f75c21bb2c44d387f8834c5c5e9c35.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/106433_f54b3c4e49864b0692e6eb2f55b8923a.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/106667_8f1126b0213a4beebc2bfc14d21cee66.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/106834_7f199fd87ae74435ba4dcdca23ebf329.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/107033_048d7d72a1c24405b267c60cdaeca80b.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/107242_12b6dfbe5c5345caa45838d47163ffe9.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/107440_7ebf797db6a648f0a8c34739db632148.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/107637_3684647948604137b95b5b6428df2c52.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/107844_cc575638f6144ba48893b6bf2c3cd3ca.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/108626_7320f31fc17740d6b74bb728266829d8.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/108818_823a5d8351884d18ae72f6a63ed7c417.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/109046_c285680cca0049acbdeed770af7e96f2.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/109222_71801ffca96443f9a8ca958cecd5d9b6.jpg\"," +
                "\"http://110.41.10.45:8001/images/2025/11/06/109420_c1eba6a051674d74b8226cacb726bf11.jpg\"," +
                "\"\"," +
                "\"\"," +
                "\"\"]", String.class).stream().filter(StrUtil::isNotEmpty).toList();

        String bucket = "baoyierp";
        String prefix = "pressure-test/";

        int threadCount = 10;   // 并发线程数
        int totalTasks = 50;    // 总任务数（会循环使用URL）

        // ====================== 初始化 ======================
        log.info("开始压测: 并发={} 总任务数={}", threadCount, totalTasks);
        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
        CountDownLatch latch = new CountDownLatch(totalTasks);
        AtomicInteger successCount = new AtomicInteger();
        AtomicInteger failCount = new AtomicInteger();
        long start = System.currentTimeMillis();

        // ====================== 并发上传 ======================
        for (int i = 0; i < totalTasks; i++) {
            final int index = i;
            executor.submit(() -> {
                String url = null;
                try {
                    // 随机或轮询一个URL
                    url = fileUrls.get(index % fileUrls.size());
                    String key = prefix + "img_" + index + "_" + System.currentTimeMillis() + ".jpg";

                    // ✅ 调用你原有的下载+上传方法
                    R<UploadFIleResultDTO> result = getUploadFIleResultDTOR(url, bucket, key, 800000L);

                    if (result.getCode() == 0 && result.getData() != null) {
                        successCount.incrementAndGet();
                        log.info("✅ 上传成功: {} -> {}", url, result.getData().getUrl());
                    } else {
                        failCount.incrementAndGet();
                        log.warn("❌ 上传失败 [{}]: {}", result.getCode(), result.getMsg());
                    }

                } catch (Exception e) {
                    failCount.incrementAndGet();
                    log.error("❌ 上传异常: url={}", url, e);
                } finally {
                    latch.countDown();
                }
            });
        }

        // ====================== 等待完成 ======================
        try {
            latch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        executor.shutdown();

        // ====================== 汇总结果 ======================
        long elapsed = System.currentTimeMillis() - start;
        double avg = elapsed / (double) totalTasks;

        log.info("""
                        ========= 压测完成 =========
                        总任务数: {}
                        成功数: {}
                        失败数: {}
                        总耗时: {} ms
                        平均单次耗时: {} ms
                        平均吞吐: {:.2f} 次/秒
                        ==========================
                        """,
                totalTasks, successCount.get(), failCount.get(),
                elapsed, String.format("%.2f", avg),
                (totalTasks * 1000.0 / elapsed)
        );
    }


    /**
     * 批量生成预签名上传 URL
     */
    public List<PresignedUrlResponse> generateBatchPresignedUploadUrls(ILoginInfo loginInfo, FileUploadBatchRequest request) {
        this.generateBatchPresignedUploadUrlsBuildLogin(loginInfo, request);

        List<FileUploadDTO> dtos = request.getDtos();
        BeanMergeUtils.mergeFillNull(request.getCommon(), dtos, fileUploadDTOMapper::update, FileUploadDTO::new);

//        Long userId = request.getUserId();
//        Long parentId = request.getParentId(); // 上传到哪个目录
//        List<FileUploadItem> files = request.getFiles();
//        String bucketName = storageService.getBucketName(null);
        String bucketName = fileStorageProperties.getBusinessBucketMap().get(EnumFileStorageBusiness.userFileStorage.name());

        List<Long> parentIds = dtos.stream().map(FileUploadDTO::getParentId).filter(NumUtils::gt0).toList();
        Map<Long, UserFileNode> parentMap = userFileNodeService.idMap(parentIds);
        AssertUtil.notEmpty(parentIds.size() == parentMap.size(), "目录异常");
        AssertUtil.isTrue(dtos.stream().allMatch(dto -> 0L == dto.getParentId() || parentMap.get(dto.getParentId()).getUserId().equals(dto.getUserId())), "目录异常");

        List<UserFileNode> fileNodes = dtos.stream().map(dto -> {
            UserFilePathParam req = new UserFilePathParam();
            req.setModuleName(UserFileCreateService.MODULE_NAME_HOME);
            req.setModuleNameTranslate(UserFileCreateService.MODULE_NAME_HOME_TRANSLATE);
            req.setUserId(dto.getUserId());
//    req.setBatchId("");
//    req.setUuid("");
            req.setOriginalFilename(dto.getNodeName());
//    req.setTemplatePhysical(dto.get);
//    req.setTemplateLogical("");
//    req.setUrl("");
            req.setCreateBy(dto.getCreatedBy());
            UserFileNode userFileNode = userFileUnitService.buildFull(req, bucketName);
            userFileNode.getUserFile().setStatus(EnumFileStatus.UPLOADING);
            return userFileNode;
        }).toList();

        userFileUnitService.saveBatchFull(fileNodes);

        return fileNodes.stream().map(fileNode -> {
            UserFile userFile = fileNode.getUserFile();
            return getPresignedUrlResponse(fileNode, userFile, false);
        }).toList();
    }

    private PresignedUrlResponse getPresignedUrlResponse(UserFileNode fileNode, UserFile userFile, boolean needCallback) {
        if (needCallback) {
            return storageService.generatePresignedUploadUrlWithCallback(userFile.getBucket(), userFile.getStorageKey(), 20 * 60, Map.of("userId", userFile.getUserId() + ""));
        } else {
            return storageService.generatePresignedUploadUrl(userFile.getBucket(), userFile.getStorageKey(), 20 * 60, fileNode.getContentType());
        }
    }

    protected void generateBatchPresignedUploadUrlsBuildLogin(ILoginInfo loginInfo, FileUploadBatchRequest request) {

    }

    private String getExtension(String fileName) {
        int idx = fileName.lastIndexOf(".");
        return idx > 0 ? fileName.substring(idx + 1) : "";
    }

    public List<UserUploadRecord> generatePresignedUploadUrls(Long userId, UserUploadRecordRequest request, String business) {
        List<UserUploadRecord> records = request.getDtos();
        UserUploadRecord common = request.getCommon();
        common.setUserId(userId);
        BeanMergeUtils.mergeFillNull(common, records, fileUploadDTOMapper::updateRecord, UserUploadRecord::new);
        String bucketName = fileStorageProperties.getBusinessBucketMap().get(business);
        records.forEach(record -> {
            String resultUrl = getResultUrl(record);
            // 获取临时上传签名
            PresignedUrlResponse response = storageService.generatePresignedUploadUrl(bucketName, resultUrl, 60 * 10L, record.getContentType());
            record.setResultUrl(resultUrl);
            record.setUploadUrl(response.getUrl());
        });
        userUploadRecordService.saveBatch(records);
        return records;
    }

    private static String getResultUrl(UserUploadRecord record) {
        String resultUrl = record.getResultUrl();
        if (StrUtil.isNotBlank(resultUrl)) {
            return resultUrl;
        }
        String originUrl = StrUtils.or(record.getOriginUrl(), "");
        return record.getUserId() + "/" + IdUtil.simpleUUID() + FileUtil.getSuffix(originUrl);
    }


//    public static void main(String[] args) {
//        System.out.println(FileUtil.getSuffix("https://chat.deepseek.com/a/chat/s/e91e3506-233d-46ff-aeb0-383558060644.png?shdd=123"));
//    }
}
